﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Apache.Cassandra;
using Lucene.Net.Documents;
using Lucene.Net.Analysis;
using System.IO;
using Lucene.Net.Analysis.Tokenattributes;
using Lucene.Net.Search;
using System.Security.Cryptography;
using Lucene.Net.Index;
using Lucandra.Net;
using Lucandra.Net.Connection.Pooling;
using Lucandra.Net.Convenience;
using System.Diagnostics;
using Lucandra.Net.Connection;

namespace Lucandra.Convenience
{
    internal sealed class LucandraHelper
    {
        const string HashChars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";

        #region Fields
        
        object lockObject = new object();
        private Dictionary<byte[], Dictionary<string, List<Mutation>>> keySpaceMutations;
        SHA1 sha1;

        #endregion

        #region Properties [public]

        public string IndexName { get; private set; }
        public string KeySpace { get; private set; }

        #endregion

        #region Constructors

        public LucandraHelper(string keySpace, string indexName)
        {
            this.KeySpace = keySpace;
            this.IndexName = indexName;

            this.keySpaceMutations = new Dictionary<byte[], Dictionary<string, List<Mutation>>>();
            this.sha1 = SHA1.Create();
        }

        #endregion

        #region Methods [public]

        public void AddDocument(string documentID, Document doc, Analyzer analyzer, Similarity similarity)
        {
            // Create the document row key.
            //string documentKey = this.IndexName + Settings.Delimeter + documentID;
            string documentKey = Settings.UseKeyspacePerIndex ? documentID : LucandraDocument.CreateDocumentKey(this.IndexName, documentID);

            // Retrieve the current mutation buffer for the "Documents" column family and the document row key.
            IList<Mutation> documentsMutationBuffer;
            lock (this.lockObject)
                documentsMutationBuffer = this.GetMutationsForColumnFamilyAndKey(Settings.DocumentsColumnFamily, documentKey);

            // Create the stored-fields supercolumn mutation (only used if fields are stored).
            var storedFieldsMutation = new SuperColumnMutation(Settings.StoredFieldsColumn);

            // Create the indexed-terms supercolumn mutation.  This stores frequency and position information next to each indexed term.
            var indexedTermsMutation = new SuperColumnMutation(Settings.IndexedTermsColumn);

            // This is used later on for indexed documents to track term position vectors.
            int position = 0;

            // This is used to keep track of how many times a field appears with a stored value (multi-valued fields) so that
            // we can assign the field in the storedFields super column a number.
            var fieldValueCount = new Dictionary<string, int>();

            var lucandraDocument = new LucandraDocument(this.IndexName, documentID);
            var termInfoDocument = new LucandraTermInfo.LucandraTermInfoDocument(documentID);

            // Iterate through all fields in the document.
            foreach (Fieldable field in doc.GetFields())
            {
                // If the field is indexed but tokenized, then we'll index the entire text of the field.
                if (field.IsIndexed() && !field.IsTokenized())
                {
                    var termInfo = new LucandraTermInfo(this.IndexName, field.Name(), field.StringValue());
                    termInfo.AddTermInfoDocument(termInfoDocument);
                    this.AddTermInfo(termInfo);
                }
                // Else, if the field is indexed and tokenized then we'll split up the terms.
                else if (field.IsIndexed())
                {
                    // Get the tokens for this document.
                    TokenStream tokens = field.TokenStreamValue();
                    if (tokens == null)
                        tokens = analyzer.TokenStream(field.Name(), new StringReader(field.StringValue()));

                    // Dictionaries to keep track of information per-term for this field for both [TermInfo] and [Documents].
                    var fieldTermInfo = new Dictionary<string, LucandraTermInfo>();
                    var indexedTermInfo = new Dictionary<string, LucandraDocument.IndexedTerm>();

                    // Initialize the offset and positions.
                    int lastOffset = 0;
                    if (position > 0)
                        position += analyzer.GetPositionIncrementGap(field.Name());

                    // Reset the token stream to the first token.
                    tokens.Reset();

                    // Create the offset attribute if the offset is stored with the term vector.
                    OffsetAttribute offsetAttribute = null;
                    if (field.IsStoreOffsetWithTermVector())
                        offsetAttribute = (OffsetAttribute)tokens.AddAttribute(typeof(OffsetAttribute));

                    // Create the position-increment attribute if the position is stored with the term vector.
                    PositionIncrementAttribute posIncrAttribute = null;
                    if (field.IsStorePositionWithTermVector())
                        posIncrAttribute = (PositionIncrementAttribute)tokens.AddAttribute(typeof(PositionIncrementAttribute));

                    // Create the term-text attribute.
                    TermAttribute termAttribute = (TermAttribute)tokens.AddAttribute(typeof(TermAttribute));

                    // Counter to keep track of the number of tokens in this field.
                    int tokensInField = 0;

                    while (tokens.IncrementToken())
                    {
                        // Increment the field token count.
                        tokensInField++;

                        string termText = termAttribute.Term();
                        string fieldName = field.Name();

                        // Create the term key.
                        string termKey = LucandraTermInfo.CreateTermKey(this.IndexName, fieldName, termText, Settings.UseKeyspacePerIndex);
                        string indexedTermKey = LucandraTermInfo.CreateTermString(fieldName, termText);

                        // Get or create the [TermInfo] container for this term.
                        LucandraTermInfo termInfo;
                        if (!fieldTermInfo.TryGetValue(termKey, out termInfo))
                        {
                            termInfo = new LucandraTermInfo(this.IndexName, fieldName, termText);
                            termInfo.AddTermInfoDocument(termInfoDocument);
                            fieldTermInfo[termKey] = termInfo;
                        }

                        // Get or create the [Documents => indexedTerms] container for this term.
                        LucandraDocument.IndexedTerm indexedTerm;
                        if (!indexedTermInfo.TryGetValue(indexedTermKey, out indexedTerm))
                        {
                            indexedTerm = new LucandraDocument.IndexedTerm(fieldName, termText, field.IsStorePositionWithTermVector(), field.IsStoreOffsetWithTermVector());
                            indexedTermInfo[indexedTermKey] = indexedTerm;
                        }

                        // Increment the term frequency.
                        termInfo.IncrementFrequency();
                        indexedTerm.IncrementFrequency();
                        termInfoDocument.IncrementFrequency();

                        // Term position vectors.
                        if (field.IsStorePositionWithTermVector())
                        {
                            position += (posIncrAttribute.GetPositionIncrement());

                            indexedTerm.AddPosition(position);
                            termInfoDocument.AddPosition(position);

                            termInfoDocument.StorePositions = true;
                        }

                        // Term offset vectors.
                        if (field.IsStoreOffsetWithTermVector())
                        {
                            int startOffset = lastOffset + offsetAttribute.StartOffset();
                            int endOffset = lastOffset + offsetAttribute.EndOffset();

                            indexedTerm.AddOffset(startOffset);
                            indexedTerm.AddOffset(endOffset);

                            termInfoDocument.AddOffset(startOffset);
                            termInfoDocument.AddOffset(endOffset);

                            termInfoDocument.StoreOffsets = true;
                        }
                    }

                    // Normalizations.
                    byte? norm = null;
                    if (!field.GetOmitNorms())
                    {
                        float boost = doc.GetBoost();
                        boost *= field.GetBoost();
                        boost *= similarity.LengthNorm(field.Name(), tokensInField);
                        norm = Similarity.EncodeNorm(boost);
                    }
                    termInfoDocument.SetNormalizationFactor(norm);

                    // Iterate through all TermInfo containers and append an insert to the mutation buffer for [TermInfo].
                    foreach (var item in fieldTermInfo)
                        this.AddTermInfo(item.Value);

                    // Iterate through all LucandraDocument.IndexedTerm containers and append an insert to the mutation buffer for [Documents].
                    // This is done on a per-field basis since a field value can contain many terms.
                    foreach (var item in indexedTermInfo)
                    {
                        var column = CassandraHelper.CreateCassandraColumn(item.Value.GetIndexedTermKey(), item.Value.GetValueForStorage());
                        indexedTermsMutation.Columns.Add(column);
                    }
                }

                // If the field value is stored, then store the field in the 'storedFields' supercolumn under the document.
                if (field.IsStored())
                {
                    // Increment or set the current stored value count for this field.
                    if (fieldValueCount.ContainsKey(field.Name()))
                        fieldValueCount[field.Name()]++;
                    else
                        fieldValueCount[field.Name()] = 0;

                    byte[] value;

                    // Create the value, depending on whether or not the field is binary.  If the field is binary, the first
                    // byte is set to '1', else if the field is not binary the first byte is set to '0'.
                    if (field.IsBinary())
                    {
                        value = new byte[field.GetBinaryLength() + 1];
                        value[0] = 1;
                        Array.Copy(field.BinaryValue(), 0, value, 1, value.Length - 1);
                    }
                    else
                    {
                        value = new byte[Encoding.UTF8.GetByteCount(field.StringValue()) + 1];
                        value[0] = 0;
                        Array.Copy(Encoding.UTF8.GetBytes(field.StringValue()), 0, value, 1, value.Length - 1);
                    }

                    // Create the Cassandra column for the mutation.
                    var column = CassandraHelper.CreateCassandraColumn(field.Name() + Settings.Delimeter + fieldValueCount[field.Name()], value);

                    // Add the column to the mutation.
                    storedFieldsMutation.Columns.Add(column);
                }
            }

            lock (this.lockObject)
            {
                documentsMutationBuffer.Add(storedFieldsMutation);
                documentsMutationBuffer.Add(indexedTermsMutation);
            }
        }

        public void AddTermInfo(LucandraTermInfo termInfo)
        {
            // Get the term row key.
            string termKey = termInfo.GetTermKey(Settings.UseKeyspacePerIndex);

            // Retrieve the current mutation buffer for the "TermInfo" column family and the term row key.
            var termInfoMutationBuffer = this.GetMutationsForColumnFamilyAndKey(Settings.TermInfoColumnFamily, termKey);

            // Iterate through all of the documents in the lucandra term info and add its entry.
            foreach (var termDocument in termInfo.Documents)
            {
                // Get the value for storage.  This is in the format [frequency][positions][offsets].
                var value = termDocument.GetValueForStorage();

                var mutation = new StandardColumnMutation(termDocument.DocumentID, value);

                // Create the cassandra column for the mutation [TermInfo => documents].
                //var column = CassandraHelper.CreateCassandraColumn(termDocument.DocumentID, value);

                // Add the column to the mutation.
                //mutation.Columns.Add(column);

                // Add the mutation to the mutation buffer.
                termInfoMutationBuffer.Add(mutation);
            }
        }

        /// <summary>
        /// Execute all of the current pending mutations in the mutation queue.
        /// </summary>
        /// <param name="client">The client on which to execute the mutations.</param>
        /// <param name="consistencyLevel">The Cassandra cluster consistency level.</param>
        public void BatchMutate(ConsistencyLevel consistencyLevel)
        {
            lock (this.lockObject)
            {
                int attempts = 0;
                do
                {
                    attempts++;
                    using (var client = CassandraConnectionPool.Borrow())
                    {
                        try
                        {
                            if (client.Keyspace != this.KeySpace)
                                client.SetKeyspace(this.KeySpace);

                            client.BatchMutate(this.keySpaceMutations, consistencyLevel);
                            this.keySpaceMutations = new Dictionary<byte[], Dictionary<string, List<Mutation>>>();
                            break;
                        }
                        catch (CassandraClientInvalidatedException)
                        {
                            client.Dispose();
                        }
                        catch (Exception ex)
                        {
                            if (attempts == 5)
                            {
                                Debug.WriteLine("============ Exception ============\r\n{0}\r\n============ Inner Exception ============\r\n{1}\r\n============",
                                    ex,
                                    ex.InnerException);

                                throw;
                            }
                        }
                    }
                }
                while (attempts < 5);
            }
        }

        public void DeleteLucandraDocument(LucandraDocument doc)
        {
            //string documentKey = doc.GetDocumentKey();
            string documentKey = Settings.UseKeyspacePerIndex ? doc.DocumentID : doc.GetDocumentKey();

            IList<Mutation> documentMutations;
            lock (this.lockObject)
                documentMutations = this.GetMutationsForColumnFamilyAndKey(Settings.DocumentsColumnFamily, documentKey);

            var documentDeletion = CassandraHelper.CreateDeletion();
            documentMutations.Add(new Mutation() { Deletion = documentDeletion });
            
            foreach (var indexedTerm in doc.IndexedTerms)
            {
                //string termInfoKey = this.IndexName + Settings.Delimeter + indexedTerm.GetIndexedTermKey();
                string termInfoKey = LucandraTermInfo.CreateTermKey(this.IndexName, indexedTerm.FieldName, indexedTerm.TermText, Settings.UseKeyspacePerIndex);

                IList<Mutation> termInfoMutations;
                lock(this.lockObject)
                    termInfoMutations = this.GetMutationsForColumnFamilyAndKey(Settings.TermInfoColumnFamily, termInfoKey);

                var slicePredicate = CassandraHelper.CreateSlicePredicate(new string[] { doc.DocumentID });
                //var termInfoDocumentDeletion = CassandraHelper.CreateDeletion(Encoding.UTF8.GetBytes(Settings.TermDocumentsColumn), slicePredicate);
                var termInfoDocumentDeletion = CassandraHelper.CreateDeletion(slicePredicate);

                lock(this.lockObject)
                    termInfoMutations.Add(new Mutation() { Deletion = termInfoDocumentDeletion });
            }
        }

        #endregion
        #region Methods [private]

        /// <summary>
        /// Get the current working mutation buffer for the specified column family and row key.
        /// </summary>
        /// <param name="columnFamily">The column family to which the key is a child.</param>
        /// <param name="key">The key (row identifier) to retrieve mutations for.</param>
        /// <returns>An IList containing the queued mutations for the specified column family and row key.</returns>
        private IList<Mutation> GetMutationsForColumnFamilyAndKey(string columnFamily, string key)
        {
            var keyBytes = Encoding.UTF8.GetBytes(key);

            List<Mutation> mutationList;
            Dictionary<string, List<Mutation>> keyMutations;

            if (!this.keySpaceMutations.TryGetValue(keyBytes, out keyMutations))
            {
                keyMutations = new Dictionary<string, List<Mutation>>();
                mutationList = new List<Mutation>();
                keyMutations.Add(columnFamily, mutationList);
                this.keySpaceMutations.Add(keyBytes, keyMutations);
                return mutationList;
            }

            if (!keyMutations.TryGetValue(columnFamily, out mutationList))
            {
                mutationList = new List<Mutation>();
                keyMutations.Add(columnFamily, mutationList);
            }

            return mutationList;
        }

        #endregion
        #region Methods [public] [static]

        public static string CreateHash(byte[] b)
        {
            StringBuilder sb = new StringBuilder(b.Length);
            int index;
            int hashCharsLength = HashChars.Length;
            bool even;

            for (int i = 0, len = b.Length; i < len; i++)
            {
                even = i % 2 == 0;
                index = b[i];
                while (index >= hashCharsLength)
                    index -= (even ? hashCharsLength : 39);
                while (index < 0)
                    index += (even ? 7 : 31);
                sb.Append(HashChars[index]);
            }

            return sb.ToString();
        }

        #endregion
    }
}
